// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/mexc/feed_v3/processor.h"

#include <set>
#include <string_view>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/mexc/feed_v3/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::mexc::feed_v3::impl {
void MexcFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  const auto& data = packet.payload;
  rapidjson::Document document;
  document.Parse(data.c_str());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  const auto& channel = ExtractChannel(document);
  if (!CheckChannelFilter(channel)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }
  if (document.HasMember("c")) {
    const std::string& c = document["c"].GetString();
    std::vector<std::string> comps = Split(c, "@");
    CHECK_GE(comps.size(), 3);
    const std::string& native_symbol = comps[2];
    int64_t ts = document["t"].GetInt64() * 1'000'000LL;

    if (channel == "snapshot") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[native_symbol];
        if (!stat.snapshot_received || !stat.update_received) {
          MexcFeedUpdateParser::ParseSnapshot(book, document["d"], ts);
          upd_mgr_.PublishBook();
          stat.snapshot_received = true;
        }
      }
    } else if (channel == "diff") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        auto& stat = topic_stat_[native_symbol];
        if (stat.snapshot_received) {
          if (parser_config_.publish_bbo_book_only()) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            bool book_generated = MexcFeedUpdateParser::ParseDiff(book, document["d"], ts);
            std::optional<PriceQty> bid_post = book->Bid0();
            std::optional<PriceQty> ask_post = book->Ask0();
            bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
            if (book_generated && is_bbo_changed) {
              upd_mgr_.PublishBook();
            }
          } else {
            MexcFeedUpdateParser::ParseDiff(book, document["d"], ts);
            upd_mgr_.PublishBook();
          }
          stat.update_received = true;
        }
      }
    } else if (channel == "ticker") {
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        MexcFeedUpdateParser::ParseTicker(book, document["d"], ts);
        upd_mgr_.PublishBook();
      }
    } else if (channel == "trade") {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};
      for (auto& trade_json : document["d"]["deals"].GetArray()) {
        MexcFeedUpdateParser::ParseTrade(trade_json, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else {
      THROW() << "[Mexc Feed Processor] unkown topic: " << packet.payload;
    }
  } else if (channel == "kline_rest") {
    const std::string& url = document["url"].GetString();
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    std::string native_symbol = parameters["symbol"];
    std::string kline_interval = parameters["interval"];
    auto interval = MexcFeedUpdateParser::ParseKlineInterval(kline_interval);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      const auto& data = document["data"].GetArray();
      int page = -1;
      if (parameters.count("page") > 0) {
        page = std::stoi(parameters["page"]);
      }
      if (page == 0) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      for (const auto& kline_data : data) {
        bool freezed = MexcFeedUpdateParser::ParseKlineRest(kline_data,
            klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else {
    THROW() << "[Mexc Feed Processor] meta message: " << packet.payload;
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::mexc::feed_v3::impl
